package com.surfilter.massdata.spark.output;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.Partitioner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.output.ISparkOutput;
import com.surfilter.massdata.spark.bean.ExDomain;
import com.surfilter.massdata.spark.util.BaseRedisDao;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.URLUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;

/**
 * 写泛域名：
 * */
public class EtlDomainOutput implements ISparkOutput {
	@Override
	public void write(DataFrame result, Map<String, String> commandMap) {
		try {
			SysLog.info("*****************EtlDomainOutput************************");

			// 转成 key:<dip1> value:<domaim>
			// key:<dip2> value:<domaim>
			JavaPairRDD<String, String> exd1 = result.toJavaRDD()
					.flatMapToPair(new transIPKeyFloatMap());

			// countByKey
			Map<String, Object> exd_count = exd1.countByKey();
			System.out.println("exd_count:"+exd_count.toString());
			Iterator it = exd_count.keySet().iterator();
			int judgeNum = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.judge.num", "200"));
			System.out.println("judgeNum:"+judgeNum);
			Map<String, Object> exdmap = new HashMap<String, Object>();
			// 过滤掉小于阀值的key
			while (it.hasNext()) {
				String key = it.next().toString();
				int num = Integer.valueOf(exd_count.get(key).toString());
				if (num > judgeNum) {
					exdmap.put(key, exd_count.get(key).toString());
				}
			}

			// 从数据集里过滤掉小于阀值的泛域名
			JavaPairRDD<String, String> exd2 = exd1.filter(new exdFilter_(
					exdmap));
			System.out.println("exd2:"+exd2.collectAsMap());
			// key:<dip1> value:<domaim> 转为 key:<domaim> value:<dip1>
			JavaPairRDD<String, String> exd3 = exd2
					.mapToPair(new transDomainKeyMapToPair());
			System.out.println("exd3:"+exd3.collectAsMap());
			// key:<domaim> value:<dip1,dip2>
			JavaPairRDD<String, String> exd4 = exd3.reduceByKey(new ipMerge());
			// 数据写入redis
			System.out.println("exd4:"+exd4.collectAsMap());
			int maxLimit = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.max.limit", "10000"));			
			exd4.foreachPartition(new FlatMapExDomain(maxLimit));
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

class transIPKeyFloatMap implements PairFlatMapFunction<Row, String, String> {
	String URL_PREFIX_HTTP = "http://";

	@Override
	public Iterable<Tuple2<String, String>> call(Row line) throws Exception {
		List<Tuple2<String, String>> results = new ArrayList<>();
		String domain = line.getAs("domain");
		String ips = line.getAs("dip");
		String[] ip_list = StringUtils.split(ips,"\\|");

		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		topDomain = topDomain.toLowerCase();

		for (String ip : ip_list) {
			results.add(new Tuple2<String, String>(ip+"\t"+topDomain, "1"));
		}

		return (Iterable<Tuple2<String, String>>) results;
	}
}

class exdFilter_ implements Function<Tuple2<String, String>, Boolean> {
	Map<String, Object> exd_count = null;

	public exdFilter_(Map<String, Object> exd_count) {
		this.exd_count = exd_count;
	}

	@Override
	public Boolean call(Tuple2<String, String> v1) throws Exception {
		String key = v1._1();
		if (exd_count.containsKey(key)) {
			return true;
		} else {
			return false;
		}
	}

}

class transDomainKeyMapToPair implements
		PairFunction<Tuple2<String, String>, String, String> {

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String[] key = StringUtils.split(t._1(), "\t");
		String ip = key[0];
		String domain = key[1];
		return new Tuple2<String, String>(domain, ip);
	}

}

class ipMerge  implements Function2<String, String, String>{

	@Override
	public String call(String v1, String v2) throws Exception {
		if(v1.contains(v2)){
			return v1;
		}else if(v2.contains(v1)){
			return v2;
		}else{
			return v1+","+v2;
		}
			
		
	}
	
}


class FlatMapExDomain  implements VoidFunction<Iterator<Tuple2<String, String>>> {
	int maxLimit=10000;
	FlatMapExDomain(int maxLimit){
		this.maxLimit=maxLimit;
	}
	@Override
	public void call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		BaseRedisDao redisDao =new BaseRedisDao();
		ObjectMapper mapper=new ObjectMapper();
		ObjectWriter writer=mapper.writer();
		
		while(t.hasNext()){
			// key:<domaim> value:<dip1,dip2>
			Tuple2<String, String> tt=t.next();
			String domain =tt._1();
			String value=tt._2();

			ExDomain exDomain = new ExDomain();
			//当前数据找到的域名、ip
			String[] value_list = StringUtils.split(value,",");
			//redis 里存的域名、ip
			String redis_value = redisDao.hgetExDomain(EX_DOMAIN_REDIS_KEY, domain);
			Set  ipSet = new HashSet();
			String ips_new = "";
			if(redis_value!=null){
				ExDomain redis_domain = mapper.readValue(StringUtil.toLowerJson(redis_value),
						ExDomain.class);
				String redis_ips = redis_domain.getIps();
				String[] ip_list = StringUtils.split(redis_ips,",");
				for (String i : ip_list) {
					 ipSet.add(i.trim());
				}
			}
			if(ipSet.contains(EX_DOMAIN_DEFAULT_IP)){
				
			}else{
				for (String i : value_list) {
					 ipSet.add(i.trim());
				}

				if ( ipSet.size() > maxLimit) {
					ips_new = EX_DOMAIN_DEFAULT_IP;
				} else {
					Iterator iterator=ipSet.iterator();
					while(iterator.hasNext()){
						ips_new+=iterator.next();
						ips_new+=",";
					}
					ips_new = ips_new.substring(0, ips_new.length() - 1);
				}
				
				exDomain.setDomain("www." + domain);
				exDomain.setIps(ips_new);
				exDomain.setSuffix(domain);
				exDomain.setFindtime(new SimpleDateFormat("yyyyMMdd")
						.format(new Date()));
				System.out.println("redis:"+StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
				redisDao.hsetExDomain(EX_DOMAIN_REDIS_KEY, domain,
						StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
			}
		}
	}
}
